﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;

namespace Util.Observable
{
    public class Observable<T> : IObservable<T>
    {
        #region Class Unsubscriber

        public class Unsubscriber : IDisposable
        {
            private HashSet<IObserver<T>> observers;
            private IObserver<T> observer;
            private ReaderWriterLock rwlock;

            public Unsubscriber(HashSet<IObserver<T>> observers, IObserver<T> observer, ReaderWriterLock rwlock)
            {
                this.observers = observers;
                this.observer = observer;
                this.rwlock = rwlock;
            }

            public void Dispose()
            {
                if (this.observer != null)
                {
                    this.rwlock.AcquireWriterLock(Timeout.Infinite);
                    observers.Remove(observer);
                    this.rwlock.ReleaseWriterLock();
                }
            }
        }

        #endregion

        private HashSet<IObserver<T>> observers;
        private ReaderWriterLock rwlock;

        public Observable()
        {
            this.observers = new HashSet<IObserver<T>>();
            this.rwlock = new ReaderWriterLock();
        }

        #region Method Notify
        public void OnCall(T obj)
        {
            rwlock.AcquireReaderLock(Timeout.Infinite);
            foreach (IObserver<T> observer in observers)
            {
                try
                {
                    observer.OnNext(obj);
                }
                catch (Exception e)
                {
                    Console.Error.WriteLine(e.ToString());
                }
            }
            rwlock.ReleaseReaderLock();
        }

        public void OnError(Exception error)
        {
            rwlock.AcquireReaderLock(Timeout.Infinite);
            foreach (IObserver<T> observer in observers)
            {
                try
                {
                    observer.OnError(error);
                }
                catch (Exception e)
                {
                    Console.Error.WriteLine(e.ToString());
                }
            }
            rwlock.ReleaseReaderLock();
        }

        public void OnCompleted()
        {
            rwlock.AcquireReaderLock(Timeout.Infinite);
            foreach (IObserver<T> observer in observers)
            {
                try
                {
                    observer.OnCompleted();
                }
                catch (Exception e)
                {
                    Console.Error.WriteLine(e.ToString());
                }
            }
            rwlock.ReleaseReaderLock();
        }
        #endregion

        public IDisposable Subscribe(IObserver<T> observer)
        {
            this.rwlock.AcquireWriterLock(Timeout.Infinite);
            if (!this.observers.Contains(observer))
            {
                this.observers.Add(observer);
            }
            else
            {
                this.rwlock.ReleaseWriterLock();
                return null;
            }
            this.rwlock.ReleaseWriterLock();
            return new Unsubscriber(this.observers, observer, this.rwlock);
        }
    }
}
